package com.shujia.streaming

import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, SparkSession}
import org.apache.spark.streaming.{Durations, StreamingContext}
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}

//spark.default.parallelism
object ForeachRDDDemo {
  def main(args: Array[String]): Unit = {
    /**
     * Spark core:        SparkContext -> RDD
     * Spark Sql:         SparkSession -> DataFrame
     * Spark Streaming:   StreamingContext -> DStream
     */
    //this(sparkContext: SparkContext, batchDuration: Duration)

    val ss: SparkSession = SparkSession.builder()
      .master("local[*]") // 如果不给线程核数，默认只有1个
      .appName("word count")
      .getOrCreate()

    import ss.implicits._
    import org.apache.spark.sql.functions._

    val sc: SparkContext = ss.sparkContext

    //    val conf = new SparkConf()
    //    conf.setMaster("local")
    //    conf.setAppName("word count")
    //    val sc = new SparkContext(conf)

    //传入一个SparkContext对象，以及接收要处理的数据时间段
    val streamingContext = new StreamingContext(sc, Durations.seconds(5L))

    //近实时处理中，数据应当是源源不断的
    //使用端口号的数据，模拟源源不断的数据源
    //监控对应端口号中的数据
    val infoDS: ReceiverInputDStream[String] = streamingContext.socketTextStream("master", 12345)

    // hello world hello java
    //使用foreachRDD算子，使用DStream中的RDD
    infoDS.foreachRDD((rdd: RDD[String]) => {
//      println("===================================")
      /**
       * 纯RDD操作，计算单词统计
       */
      //      rdd.flatMap(_.split(" "))
      //        .map((_,1))
      //        .reduceByKey(_+_)
      //        .foreach(println)

      /**
       * 将RDD变成DataFrame做操作
       *
       * 将来就可以在近实时的环境下，使用sql语句处理窗口中的数据。
       */
      val df1: DataFrame = rdd.toDF("line")
      df1.createOrReplaceTempView("wc")
      val resDF: DataFrame = ss.sql(
        """
          |
          |select
          |t1.word as word,
          |count(1) as counts
          |from
          |(
          |select
          |explode(split(line,' ')) as word
          |from
          |wc) t1
          |group by t1.word
          |""".stripMargin)
      resDF.show()
    })


    /**
     * 额外提交命令，一直启动sparkstreaming程序
     */
    streamingContext.start()
    streamingContext.awaitTermination()
    streamingContext.stop()
  }

}
